package com.kaf.ai.service.impl;

import com.fasterxml.jackson.databind.ObjectMapper;
import com.kaf.ai.config.AiConfig;
import com.kaf.ai.config.JacksonConfig;
import com.kaf.ai.constant.AiProvider;
import com.kaf.ai.dto.AiRequest;
import com.kaf.ai.dto.AiResponse;
import com.kaf.ai.dto.DeepSeekResponse;
import com.kaf.ai.dto.DeepSeekStreamResponse;
import com.kaf.ai.service.AiService;
import lombok.extern.slf4j.Slf4j;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.http.HttpHeaders;
import org.springframework.http.MediaType;
import org.springframework.stereotype.Service;
import org.springframework.web.reactive.function.BodyInserters;
import org.springframework.web.reactive.function.client.WebClient;
import reactor.core.publisher.Flux;
import reactor.core.publisher.Mono;
import reactor.core.scheduler.Schedulers;

import java.util.List;
import java.util.Map;
import java.util.Optional;

@Slf4j
@Service
public class DeepSeekAiServiceImpl implements AiService {

    private final WebClient webClient;
    private final AiConfig.AiProperties properties;
    JacksonConfig jacksonConfig = new JacksonConfig();

    public DeepSeekAiServiceImpl(WebClient webClient,
                                 @Autowired(required = false) AiConfig aiConfig) {
        this.webClient = webClient;
        this.properties = aiConfig != null ? aiConfig.deepSeekProperties() : new AiConfig.AiProperties();
    }


    public Flux<String> streamCompletion(Map<String, Object> requestBody) {
        // 启用流式
        requestBody.put("stream", true);

        return webClient.post()
                .uri(properties.getBaseUrl() + "/chat/completions")
                .header("Authorization", "Bearer " + properties.getApiKey())
                .contentType(MediaType.APPLICATION_JSON)
                .body(BodyInserters.fromValue(requestBody))
                .retrieve()
                .bodyToFlux(String.class) // 按行接收 SSE
                .subscribeOn(Schedulers.boundedElastic()) // 避免阻塞 Netty 线程
                .handle((line, sink) -> {
                    if (line == null || line.isBlank()) return;

                    // SSE 格式: data: {...}
                    if (line.startsWith("data:")) {
                        String jsonData = line.substring(5).trim(); // 去掉 "data:"

                        if ("[DONE]".equals(jsonData)) {
                            sink.complete();
                            return;
                        }

                        try {
                            DeepSeekStreamResponse response = jacksonConfig.objectMapper().readValue(jsonData, DeepSeekStreamResponse.class);
                            String content = Optional.ofNullable(response)
                                    .map(DeepSeekStreamResponse::getChoices)
                                    .filter(list -> !list.isEmpty())
                                    .map(list -> list.get(0))
                                    .map(DeepSeekStreamResponse.Choice::getDelta)
                                    .map(DeepSeekStreamResponse.Delta::getContent)
                                    .orElse("");

                            if (content != null && !content.isEmpty()) {
                                sink.next(content);
                            }

                            // 检查是否结束
                            String finishReason = Optional.ofNullable(response)
                                    .map(r -> r.getChoices().get(0))
                                    .map(DeepSeekStreamResponse.Choice::getFinishReason)
                                    .orElse(null);

                            if (finishReason != null) {
                                sink.complete();
                            }

                        } catch (Exception e) {
                            // 记录日志但不中断流（容忍个别解析失败）
                            // log.warn("解析 DeepSeek 流失败: {}", jsonData, e);
                        }
                    }
                });
    }



    @Override
    public Mono<AiResponse> generateResponse(AiRequest request) {
        log.info("调用 DeepSeek 模型: {}, prompt: {}", properties.getModel(), request.getPrompt());

        return webClient.post()
                .uri(properties.getBaseUrl() + "/chat/completions")
                .header(HttpHeaders.AUTHORIZATION, "Bearer " + properties.getApiKey())
                .header(HttpHeaders.CONTENT_TYPE, "application/json")
                .bodyValue(buildRequestBody(request))
                .retrieve()
                .bodyToMono(String.class) // 👈 先转成 String 看原始内容
                .doOnNext(json -> log.info("DeepSeek 原始响应: {}", json)) // 打印
                .flatMap(json -> {
                    try {
                        log.info("json= "+json);
                        // 手动反序列化，便于调试
                        DeepSeekResponse response = jacksonConfig.objectMapper().readValue(json, DeepSeekResponse.class);
                        log.info("反序列化成功: {}", response);
                        return Mono.just(response);
                    } catch (Exception ex) {
                        log.error("JSON 反序列化失败", ex);
                        return Mono.error(new RuntimeException("解析 DeepSeek 响应失败", ex));
                    }
                })
                .map(this::convertToAiResponse)
                .onErrorResume(e -> {
                    log.error("DeepSeek 调用失败", e);
                    return Mono.just(AiResponse.error("调用 DeepSeek 失败: " + e.getMessage(), AiProvider.DEEPSEEK.name()));
                });
    }

    private Object buildRequestBody(AiRequest request) {
        return new RequestBody(
                properties.getModel(),
                request.getMessages() != null && !request.getMessages().isEmpty() ?
                        request.getMessages() :
                        List.of(new AiRequest.Message("user", request.getPrompt())),
                request.getTemperature(),
                request.getMaxTokens()
        );
    }

    private AiResponse convertToAiResponse(DeepSeekResponse response) {
        if (response == null) {
            return AiResponse.error("DeepSeek 响应为空", AiProvider.DEEPSEEK.name());
        }

        List<DeepSeekResponse.Choice> choices = response.getChoices();
        if (choices == null || choices.isEmpty()) {
            return AiResponse.error("DeepSeek 未返回 choices", AiProvider.DEEPSEEK.name());
        }

        DeepSeekResponse.Message message = choices.get(0).getMessage();
        if (message == null) {
            return AiResponse.error("DeepSeek 未返回 message", AiProvider.DEEPSEEK.name());
        }

        String content = message.getContent();
        if (content == null) {
            content = ""; // 或返回错误
        }

        return AiResponse.success(content,"" ,AiProvider.DEEPSEEK.name());
    }

    // 内部类映射 API 响应
    private static class RequestBody {
        private String model;
        private List<AiRequest.Message> messages;
        private Double temperature;
        private Integer max_tokens;

        public RequestBody(String model, List<AiRequest.Message> messages, Double temperature, Integer max_tokens) {
            this.model = model;
            this.messages = messages;
            this.temperature = temperature;
            this.max_tokens = max_tokens;
        }
        // Getters 用于序列化
        public String getModel() { return model; }
        public List<AiRequest.Message> getMessages() { return messages; }
        public Double getTemperature() { return temperature; }
        public Integer getMax_tokens() { return max_tokens; }
    }






    /***************************************************/




}